package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSqlApp;
import com.atguigu.gmall.realtime.bean.KeywordStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.IkAnalyzer;
import com.atguigu.gmall.realtime.function.KwProductUdtf;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneOffset;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/1 14:43
 */
public class DwsProductKeywordSearch extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsProductKeywordSearch().init(4005, 1, "DwsProductKeywordSearch");
    }
    
    @Override
    protected void run(StreamTableEnvironment tenv) {
        tenv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));
        // 1. 创建动态表与source关联:kafka中的topic  dwd_page
        tenv.executeSql("create table product_stats(" +
                            "   stt string, " +
                            "   edt string, " +
                            "   sku_name string, " +
                            "   click_ct bigint, " +
                            "   order_ct bigint, " +
                            "   cart_ct bigint, " +
                            "   ts bigint " +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092', " +
                            "   'properties.group.id' = 'DwsProductKeywordSearch', " +
                            "   'topic' = '" + Constant.TOPIC_DWS_PRODUCT_STATS + "', " +
                            "   'scan.startup.mode' = 'earliest-offset', " +
                            "   'format' = 'json' " +
                            ")");
        
        // 1. 过滤出需要的数据
        Table t1 = tenv.sqlQuery("select" +
                                     " * " +
                                     "from product_stats " +
                                     "where click_ct > 0 " +
                                     "or order_ct > 0 " +
                                     "or cart_ct > 0");
        
        tenv.createTemporaryView("t1", t1);
        
        // 2. 对sku_name分词
        tenv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        
        Table t2 = tenv.sqlQuery("select" +
                                     " stt," +
                                     " edt," +
                                     " word," +
                                     " click_ct," +
                                     " order_ct," +
                                     " cart_ct," +
                                     " ts " +
                                     "from t1, lateral table(ik_analyzer(sku_name))");
        
        tenv.createTemporaryView("t2", t2);
        // 3. 按照时间和word进行聚合
        Table t3 = tenv.sqlQuery("select" +
                                     " stt," +
                                     " edt," +
                                     " word," +
                                     " sum(click_ct) click_ct, " +
                                     " sum(order_ct) order_ct, " +
                                     " sum(cart_ct) cart_ct " +
                                     "from t2 " +
                                     "group by stt, edt, word");
        tenv.createTemporaryView("t3", t3);
        
        // 4. 多列变多行多列
        tenv.createTemporaryFunction("kw_product", KwProductUdtf.class);
        Table result = tenv.sqlQuery("select" +
                                         " stt, " +
                                         " edt," +
                                         " word keyword," +
                                         " source," +
                                         " ct, " +
                                         " unix_timestamp() * 1000 ts " +
                                         "from t3 " +
                                         "join lateral table(kw_product(click_ct, order_ct, cart_ct)) on true");
        // 5. 写入到clickhouse中
        tenv
            .toRetractStream(result, KeywordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil
                         .getClickhouseSink(Constant.CLICKHOUSE_DB,
                                            Constant.CLICKHOUSE_TABLE_KEYWORD_STATS,
                                            KeywordStats.class));
    }
}
